package com.pai4j.connect.service;

import com.pai4j.common.enums.messagequeue.MessageBroadChannelEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * 异步消息处理服务
 * 用于处理WebSocket消息的异步广播，避免阻塞WebSocket连接
 *
 * @author: CYM-pai
 * @date: 2025/08/27 23:47
 **/
@Slf4j
@Service
public class AsyncMessageService {

    @Autowired
    private MessageQueueProducer<Object, Long> messageQueueProducer;

    /**
     * 异步广播消息
     *
     * @param channel 消息频道
     * @param message 消息内容
     * @param userId 用户ID（用于日志记录）
     * @param sessionId WebSocket会话ID（用于日志记录）
     * @return CompletableFuture<Boolean> 异步执行结果
     */
    @Async("messageExecutor")
    public CompletableFuture<Boolean> broadcastMessageAsync(MessageBroadChannelEnum channel, Object message, String userId, String sessionId) {
        try {
            log.debug("Starting async broadcast for userId:{}, sessionId:{}, channel:{}", userId, sessionId, channel.getChannel());
            messageQueueProducer.broadSend(channel, message);
            log.debug("Successfully completed async broadcast for userId:{}, sessionId:{}", userId, sessionId);
            return CompletableFuture.completedFuture(true);
        } catch (Exception e) {
            log.error("Failed to broadcast message asynchronously. userId:{}, sessionId:{}, channel:{}, error:{}", 
                    userId, sessionId, channel.getChannel(), e.getMessage(), e);
            return CompletableFuture.completedFuture(false);
        }
    }

    /**
     * 异步广播消息（带回调处理）
     *
     * @param channel 消息频道
     * @param message 消息内容
     * @param userId 用户ID
     * @param sessionId WebSocket会话ID
     * @param onSuccess 成功回调
     * @param onFailure 失败回调
     */
    @Async("messageExecutor")
    public void broadcastMessageWithCallback(MessageBroadChannelEnum channel, Object message, String userId, String sessionId,
                                           Runnable onSuccess, Runnable onFailure) {
        try {
            log.debug("Starting async broadcast with callback for userId:{}, sessionId:{}", userId, sessionId);
            messageQueueProducer.broadSend(channel, message);
            log.debug("Successfully completed async broadcast with callback for userId:{}, sessionId:{}", userId, sessionId);
            if (onSuccess != null) {
                onSuccess.run();
            }
        } catch (Exception e) {
            log.error("Failed to broadcast message asynchronously with callback. userId:{}, sessionId:{}, error:{}", 
                    userId, sessionId, e.getMessage(), e);
            if (onFailure != null) {
                onFailure.run();
            }
        }
    }
}
